package cn.doitedu.cn.doitedu.rtdw.dashboard;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.hive.HiveCatalog;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/8/31
 * @Desc: 学大数据，上多易教育
 * 以下3个活动页面
 * - /mall/promotion/qxlx.html
 * - /mall/promotion/qxcd.html
 * - /mall/promotion/cmjh.html
 * 当天累计到此刻的来自于各省份 pv量，uv量，新用户数（每5分钟更新一次）
 **/
public class Job4_QixiPromotionPageUvPv {
    public static void main(String[] args) {

        // 编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 切换到hiveCatalog.doit40_rtdw库

        // 构造一个hive catalog
        HiveCatalog hiveCatalog = new HiveCatalog("doit40-hive", null, "rtdw/src/main/resources/");
        tenv.registerCatalog("doit40-hive", hiveCatalog);
        //tenv.useCatalog("doit40-hive");
        //tenv.useDatabase("doit40_rtdw");

        // dash_qixi_page_pv_uv_newuv
        // 建表，映射mysql中的结果表
        tenv.executeSql(
                " create table dash_qixi_page_pv_uv_newuv_mysql(     " +
                        "    window_start timestamp(3),            " +
                        "    window_end timestamp(3),              " +
                        "    page_url STRING,                   " +
                        "    pv_cnt bigint,                        " +
                        "    uv_cnt bigint,                        " +
                        "    new_uv_cnt bigint,                        " +
                        "    PRIMARY KEY (window_start,window_end) NOT ENFORCED     " +
                        " ) with (                                      " +
                        "  'connector' = 'jdbc',                            " +
                        "  'url' = 'jdbc:mysql://doitedu:3306/doit40',  " +
                        "  'table-name' = 'dash_qixi_page_pv_uv_newuv',             " +
                        "  'username' = 'root',                             " +
                        "  'password' = 'root'                              " +
                        " )                                                 "
        );



        // 写运算sql ==> 从表： doit40-hive.doit40_rtdw.dwd_kafka 计算
        tenv.executeSql(
                "insert into dash_qixi_page_pv_uv_newuv_mysql " +
                        "WITH tmp AS (                                                                            " +
                        "  SELECT                                                                                   " +
                        "  	event_id,                                                                               " +
                        "  	page_url,                                                                               " +
                        "   user_id,                                                                                " +
                        "   if(to_date(date_format(register_time,'yyyy-MM-dd')) = CURRENT_DATE,1,0)  as is_new,     " +
                        "  	rt                                                                                      " +
                        "  from `doit40-hive`.`doit40_rtdw`.`dwd_kafka`                                             " +
                        "  where event_id = 'page_load' and page_url in (                                           " +
                        "     '/mall/promotion/qxlx.html',                                                          " +
                        "     '/mall/promotion/qxcd.html',                                                          " +
                        "     '/mall/promotion/cmjh.html'                                                           " +
                        "     )                                                                                     " +
                        "  )                                                                                        " +
                        "                                                                                           " +
                        "  SELECT                                                                                   " +
                        "   window_start,                                                                           " +
                        "  	window_end,                                                                             " +
                        "  	page_url,                                                                               " +
                        "  	count(1) as pv_cnt,                                                                     " +
                        "  	count(distinct user_id) as uv_cnt,                                                      " +
                        "  	count(1) FILTER(WHERE is_new = 1)  as new_uv_cnt                                        " +
                        "  FROM TABLE(                                                                              " +
                        "      CUMULATE(TABLE tmp, DESCRIPTOR(rt),INTERVAL '5' MINUTE,INTERVAL '24' HOUR)           " +
                        "  )                                                                                        " +
                        "  GROUP BY                                                                                 " +
                        "      window_start,                                                                        " +
                        "  	window_end,                                                                             " +
                        "  	page_url                                                                                "
        );

    }
}
